package com.navinfo.tripanalysis.offline.service;

import com.navinfo.tripanalysis.common.arithmetic.common.InterRangeData;
import com.navinfo.tripanalysis.common.arithmetic.common.OuterEventData;
import com.navinfo.tripanalysis.common.util.GsonUtils;
import com.navinfo.tripanalysis.common.util.JsonUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.client.Put;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.types.StructType;
import org.bson.Document;

import java.io.Serializable;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;

/**
 * 行程事件信息的转换服务接口
*/
public interface TripEventConvertService extends Serializable {
    /**
     * 创建Hive表的schema
     */
    StructType createSchema();

    /**
     * 通过OuterEventData对象创建hive的Row对象
     * @param m 行程事件信息
     * @param schema 对应的schema
     * @return row
     */
    Row toRow(OuterEventData m, StructType schema);

    /**
     * Hive的Row转为Pojo对象
     * @param row hive的Row对象
     * @return  行程事件信息
     */
    OuterEventData fromRow(Row row);

    /**
     * 通过OuterEventData对象创建Mongo的Document对象
     * @param outerEventData 行程事件信息
     * @return Document
     */
    Document toDocument(OuterEventData outerEventData);
    /**
     * Mongo的Document对象转为Pojo对象
     * @param doc Document对象
     * @return 行程事件信息
     */
    OuterEventData fromDocument(Document doc);

    /**
     * 通过OuterEventData对象创建HBase的Put对象
     * @param m 行程事件信息
     * @return Put
     */
    Put toPut(OuterEventData m);


    default Long getLong(Row row, String key) {
        Object o = row.getAs(key);
        if (o != null) {
            return Long.valueOf(o.toString());
        }
        return null;
    }

    default Integer getInteger(Row row, String key) {
        Object o = row.getAs(key);
        if (o != null) {
            return Integer.valueOf(o.toString());
        }
        return null;
    }
    default Double getDouble(Row row, String key) {
        Object o = row.getAs(key);
        if (o != null) {
            return Double.valueOf(o.toString());
        }
        return null;
    }

    default String getString(Row row, String key) {
        Object o = row.getAs(key);
        if (o != null) {
            return o.toString();
        }
        return null;
    }

    default Map<Integer, InterRangeData> getRangeData(Row row, String colName) {
        String jsonStr=getString(row,colName);
        if(StringUtils.isNotEmpty(jsonStr)){
            Map<String, InterRangeData> map0= GsonUtils.gsonToMap(jsonStr);
            Map<Integer, InterRangeData> map=new LinkedHashMap<>();
            for( Map.Entry<String, InterRangeData> entry:map0.entrySet()){
                map.put(Integer.valueOf(entry.getKey()),entry.getValue());
            }
            return map;
        }
        return null;
    }

    default void putRangeData(Document doc, Map<Integer, InterRangeData> ranges, String name) {
        if(ranges!=null){
            doc.put(name,Document.parse(JsonUtils.toJsonStr(ranges)));
        }
    }

    default Map<Integer, InterRangeData> getRangeData(Document doc, String name) {
        if(doc.containsKey(name)){
            Document subDoc=(Document)doc.get(name);
            Map<Integer, InterRangeData> map=new HashMap<>();
            for(Map.Entry<String, Object> entry1: subDoc.entrySet()){
                Integer key=Integer.valueOf(entry1.getKey());
                InterRangeData value= GsonUtils.gsonToBean(GsonUtils.gsonString(entry1.getValue()), InterRangeData.class);
                map.put(key,value);
            }
            return map;
        }
        return null;
    }

}
